package com.lagou.streamsink;


import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.base.VoidSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

public class Mains {
    static class ConnectionState {

        private final transient Connection connection;

        ConnectionState(Connection connection) {
            this.connection = connection;
        }

    }
    public static void main(String[] args) throws Exception {
        String url = "jdbc:mysql://localhost:3306/bigdata?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC";
        String user = "root";
        String password = "lucas";
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(1000);
        env.setParallelism(1);
//        DataStreamSource<Tuple2<String,Integer>> data = env.fromElements(new Tuple2<String,Integer>("aa",1),new Tuple2<String, Integer>("bb",2));
        Student stu1 = new Student("lucas", 18);
        Student stu2 = new Student("jack", 28);
        DataStreamSource<Student> data = env.fromElements(stu1,stu2);

//        data.print();
        data.addSink(new TwoPhaseCommitSinkFunction<Student, ConnectionState, Void>(new KryoSerializer<>(ConnectionState.class, new ExecutionConfig()), VoidSerializer.INSTANCE) {
            Connection connection = null;
            PreparedStatement preparedStatement = null;
            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("..........open");

            }

            @Override
            protected void invoke(ConnectionState transaction, Student value, Context context) throws Exception {
               /* Connection connection = transaction.connection;
                PreparedStatement pstm = connection.prepareStatement("INSERT INTO t_wordcount (word, counts) VALUES (?, ?)");
//                PreparedStatement pstm = connection.prepareStatement("INSERT INTO t_wordcount (word, counts) VALUES (?, ?) ON DUPLICATE KEY UPDATE counts = ?");
//                PreparedStatement pstm = connection.prepareStatement("INSERT INTO student (name, age) VALUES (?, ?) ON DUPLICATE KEY UPDATE counts = ?");
                pstm.setString(1, value.f0);
                pstm.setInt(2, value.f1);
//                pstm.setInt(3, value.f1);
                pstm.executeUpdate();
                pstm.close();*/
                System.out.println("...........invoke");
                preparedStatement.setString(1,value.getName());
                preparedStatement.setInt(2,value.getAge());
                preparedStatement.executeUpdate();
            }

            @Override
            protected ConnectionState beginTransaction() throws Exception {
                System.out.println("=====> beginTransaction... ");
//                Class.forName("com.mysql.cj.jdbc.Driver");
//                String url = "jdbc:mysql://localhost:3306/bigdata?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC";
//                Connection connection = DriverManager.getConnection(url, "root", "lucas");
//                connection = DruidConnectionPool.getConnection();
                connection = DriverManager.getConnection(url, user, password);
                connection.setAutoCommit(true);

                /*System.out.println(".............beginTransaction");
                connection = DriverManager.getConnection(url, user, password);*/
                String sql = "insert into student (name,age) values (?,?)";
                preparedStatement = connection.prepareStatement(sql);

                return new ConnectionState(connection);
            }

            @Override
            protected void preCommit(ConnectionState transaction) throws Exception {
                System.out.println("=====> preCommit... " + transaction);
            }

            @Override
            protected void commit(ConnectionState transaction) {
                System.out.println("=====> commit... ");
                Connection connection = transaction.connection;
                try {
                    connection.commit();
                    connection.close();
                } catch (SQLException e) {
                    throw new RuntimeException("提交事物异常");
                }
            }

            @Override
            protected void abort(ConnectionState transaction) {
                System.out.println("=====> abort... ");
                Connection connection = transaction.connection;
                try {
                    connection.rollback();
                    connection.close();
                } catch (SQLException e) {
                    throw new RuntimeException("回滚事物异常");
                }
            }
        });

        env.execute();
    }
}
